package com.ocp.socket.task.service.impl;

import com.alibaba.fastjson.JSON;
import com.ocp.socket.busi.device.service.IDeviceInfoService;
import com.ocp.socket.busi.warnItem.bean.BaseWarnItem;
import com.ocp.socket.busi.warnItem.mapper.BaseWarnItemMapper;
import com.ocp.socket.busi.warnMsg.bean.BaseWarnMsg;
import com.ocp.socket.busi.warnMsg.service.IWarnMsgService;
import com.ocp.socket.common.Constants;
import com.ocp.socket.common.bean.DevInfo;
import com.ocp.socket.common.enums.CacheKeyEnum;
import com.ocp.socket.common.utils.*;
import com.ocp.socket.task.service.ITaskService;
import org.apache.mina.core.session.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @Author : lichong
 * @description :
 * @Date Create in 12:54 2019/3/16
 * @Modified By :
 **/
@Service("taskService")
public class TaskServiceImpl implements ITaskService {
    private static final Logger logger = LoggerFactory.getLogger(TaskServiceImpl.class);
    @Autowired
    private IDeviceInfoService deviceInfoService;
    @Autowired
    private BaseWarnItemMapper baseWarnItemMapper;
    @Autowired
    private IWarnMsgService warnMsgService;
    @Autowired
    private RedisUtil redisUtil;

    /**
     * 检查过期的设备信息
     */
    @Override
    public void checkExpireDeviceInfo() {
        long time = System.currentTimeMillis();
        //服务器启动后60s内不检测该断开的心跳
        if (time - Constants.NOT_CHECK_STOP_HEART_TIME < Constants.SERVER_START_TIME) {
            return;
        }
        //获取过期的心跳信息
        List<DevInfo> devInfos = deviceInfoService.getStopHeartDevConInfoList();
        logger.debug("心跳异常的设备数量：{}", devInfos.size());
        //判断设备正常连接是否超过正常的次数，只有超过正常次数时才报警。
        Iterator<DevInfo> devInfoIterator = devInfos.iterator();
        while (devInfoIterator.hasNext()) {
            DevInfo devInfo = devInfoIterator.next();
            IoSession session = SessionUtil.getDeviceSessionBySnNo(devInfo.getSnNo());
            if (session != null) {
                int connectNum = (Integer)session.getAttribute(Constants.ATTRIBUTE_KEY_VALID_CONNECT_NUM);
                if (connectNum <= Constants.NORMAL_CONNECT_NUM) {
                    devInfoIterator.remove();
                }
                session.setAttribute(Constants.ATTRIBUTE_KEY_VALID_CONNECT_NUM, 0);
            }
        }
        if (CollectionUtils.isEmpty(devInfos)) {
            return;
        }
        logger.info("心跳异常的设备有：{}", JSON.toJSONString(devInfos));
        //获取奶厅id信息
        List<Integer> hallIds = devInfos.stream().map(DevInfo::getHallId).distinct().collect(Collectors.toList());
        //获取已配置需要报警信息
        List<BaseWarnItem> warnItems = baseWarnItemMapper.getBaseWarnItemListByHallIds(hallIds);
        Map<Integer, List<BaseWarnItem>> hallWarnMap = warnItems.stream().collect(Collectors.groupingBy(BaseWarnItem::getHallId));
        Map<Integer, List<DevInfo>> hallDevMap = devInfos.stream().collect(Collectors.groupingBy(DevInfo::getHallId));
        Iterator<Map.Entry<Integer, List<DevInfo>>> iterator = hallDevMap.entrySet().iterator();
        List<DevInfo> devInfoList = null;
        Integer hallId = null;
        List<BaseWarnItem> baseWarnItems = null;
        Map<String, BaseWarnItem> warnItemMap = null;
        List<BaseWarnMsg> warnMsgs = new ArrayList<>();
        //删除过期的设备信息
        String key = CacheKeyEnum.DEVICE_ONLINE_SET.getCacheKey();
        Map<Integer, List<String>> msgMap = new HashMap<>(16);
        String now = DateUtil.getNowStr(DateUtil.PATTERN_12);
        while (iterator.hasNext()) {
            Map.Entry<Integer, List<DevInfo>> devMap = iterator.next();
            devInfoList = devMap.getValue();
            hallId = devMap.getKey();
            //获取配置的报警信息
            baseWarnItems = hallWarnMap.get(hallId);
            //如果该奶厅没有配置该报警信息
            if (CollectionUtils.isEmpty(baseWarnItems)) {
                //删除该设备的在线信息
                redisUtil.zRem(key, devInfos.toArray(new DevInfo[devInfos.size()]));
                continue;
            }
            warnItemMap = baseWarnItems.stream().collect(Collectors.toMap(BaseWarnItem::getWarnCode, Function.identity()));
            for (DevInfo devInfo : devInfoList) {
                String warnCode = EnumUtil.getWarnCode(devInfo.getTypeCode());
                BaseWarnItem baseWarnItem = warnItemMap.get(warnCode);
                if (baseWarnItem == null) {
                    //删除该设备的在线信息
                    redisUtil.zRem(key, devInfo);
                    continue;
                }
                BaseWarnMsg warnMsg = getWarnMsg(baseWarnItem, devInfo, now);
                warnMsgs.add(warnMsg);
                getSendMsg(warnMsg, msgMap);
            }
        }
        logger.debug("长时间未心跳,需要推送的预警信息：{}", JSON.toJSONString(warnItems));
        redisUtil.zRem(key, devInfos.toArray(new DevInfo[devInfos.size()]));
        if (CollectionUtils.isEmpty(warnMsgs)) {
            return;
        }
        //推送预警信息
        sendMsg(msgMap);
        //保存报警信息
        warnMsgService.saveWarnMsg(warnMsgs);
        for (BaseWarnMsg warnMsg : warnMsgs) {
            //放入消息队列
            redisUtil.rPush(CacheKeyEnum.WARN_MSG_PUSH_LIST.getCacheKey(), JSON.toJSONString(warnMsg));
        }
    }

    /**
     *  发送消息
     * @param msgMap
     */
    private void sendMsg(Map<Integer, List<String>> msgMap) {
        Iterator<Map.Entry<Integer, List<String>>> iterator = msgMap.entrySet().iterator();
        List<IoSession> sessions = null;
        List<String> msgs = null;
        Integer orgId = null;
        while (iterator.hasNext()) {
            Map.Entry<Integer, List<String>> next = iterator.next();
            orgId = next.getKey();
            msgs = next.getValue();
            sessions = SessionUtil.getAppSessionsByOrgId(orgId);
            if (CollectionUtils.isEmpty(sessions)) {
                continue;
            }
            for (IoSession session : sessions) {
                for (String msg : msgs) {
                    SessionUtil.sendAppMsg(session, msg);
                }
            }
        }
    }

    /**
     * 获取报警信息
     * @param item      报警配置信息
     * @param devInfo   设备信息
     */
    public BaseWarnMsg getWarnMsg(BaseWarnItem item, DevInfo devInfo, String now) {
        BaseWarnMsg warnMsg = new BaseWarnMsg();
        warnMsg.setId(CommonUtils.getUUID());
        warnMsg.setWarnCode(item.getWarnCode());
        warnMsg.setSnNo(devInfo.getSnNo());
        warnMsg.setWarnName(item.getWarnName());
        warnMsg.setWarnNameEn(item.getWarnNameEn());
        warnMsg.setMilkHallId(devInfo.getHallId());
        warnMsg.setDataTime(now);
        warnMsg.setStartTime(now);
        warnMsg.setOrgId(devInfo.getOrgId());
        warnMsg.setWarnContent(item.getWarnContent());
        warnMsg.setWarnContentEn(item.getWarnContentEn());
        warnMsg.setWarnReason(item.getWarnReason());
        warnMsg.setWarnReasonEn(item.getWarnReasonEn());
        return warnMsg;
    }

    /**
     * 获取推送的信息
     * @param warnMsg   预警信息
     * @param msgMap
     */
    public void getSendMsg(BaseWarnMsg warnMsg, Map<Integer, List<String>> msgMap) {
        List<String> strings = msgMap.get(warnMsg.getOrgId());
        String msg = CommonUtils.toMsg("0006", warnMsg.getSnNo(),warnMsg.getDataTime(),
                JSON.toJSONString(warnMsg));
        if (strings == null) {
            strings = new ArrayList<>();
            strings.add(msg);
            msgMap.put(warnMsg.getOrgId(), strings);
        } else {
            strings.add(msg);
        }

    }
}
